package com.batyuta.sort;

import org.apache.log4j.Logger;

import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;

public abstract class AbstractMergeThread<T extends MergeQueueItem> implements Runnable {
    private static int taskCount = 0;
    private final int id = taskCount++;

    volatile Queue<T> queueIn;
    volatile Queue<T> queueOut;
    private final CountDownLatch startSignal;
    private final CountDownLatch doneSignal;
    protected T currentQueue = null;

    protected AbstractMergeThread(Queue<T> queueIn, Queue<T> queueOut, CountDownLatch startSignal, CountDownLatch doneSignal) {
        this.queueIn = queueIn;
        this.queueOut = queueOut;
        this.startSignal = startSignal;
        this.doneSignal = doneSignal;
    }

    public static String getNextString(Iterator<String> iterator) {
        String result = null;
        try {
            if (iterator.hasNext()) {
                result = iterator.next();
            }
        } catch (NoSuchElementException e) {
            //nothing
        }
        return result;
    }

    @Override
    public void run() {
        try {
            if (null != startSignal) {
                startSignal.await();
            }
            process();
        } catch (InterruptedException e) {
            getLogger().error(e, e);
        } finally {
            if (null != doneSignal) {
                doneSignal.countDown();
            }
        }
    }

    public abstract T process();

    public Integer getId() {
        return id;
    }

    protected T poll() {
        T result = null;
        while (!queueIn.isEmpty() && null == (result = queueIn.poll())) {
            Thread.yield();
        }
        return result;
    }

    protected void add(T t) {
        queueOut.add(t);
    }

    abstract Logger getLogger();

    protected void addToQueue(T item) {
        if (null != currentQueue) {
            currentQueue.setFileSecond(item.getFileOut());
            add(currentQueue);
            currentQueue = null;
        } else {
            currentQueue = (T) T.createNewFromResult(item);
        }
    }
}